package com.teemor.im.executor;

import com.teemor.im.common.command.EventType;
import com.teemor.im.common.command.RequestType;
import com.teemor.im.common.command.RemoteCommand;
import com.teemor.im.common.model.ImMessage;
import com.teemor.im.common.util.JsonUtil;
import com.teemor.im.executor.handler.CommandHandler;
import com.teemor.im.discovery.node.LocalCacheNodeTable;
import com.teemor.im.discovery.node.ServerNode;
import com.teemor.im.remoting.instance.NettyRemotingClient;
import com.teemor.im.remoting.netty.ResponseFuture;
import com.teemor.im.remoting.window.SessionWindow;
import com.teemor.im.remoting.window.WindowManager;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 命令调度分发器
 *
 * @ename zhoulk
 * @cname 周李科
 * @date 2020/6/17 14:31
 */
@Slf4j
public class CommandDispatcher {

    private ThreadPoolExecutor taskExecutor;

    private ExecutorService loopThread;

    private CommandQueue commandQueue = new CommandQueue();

    private static final CommandDispatcher instance = new CommandDispatcher();

    private static final Map<EventType, CommandHandler> handlerMap;


    private static final CommandHandler newWindowHandler = (task) -> {
        RemoteCommand command = task.getCommand();
        String content = command.getContent();
        SessionWindow window = JsonUtil.jsonToPojo(content, SessionWindow.class);
        WindowManager.addWindow(window);

        RemoteCommand respCommand = new RemoteCommand();
        command.setEventType(EventType.OK);
        return respCommand;
    };

    private static final CommandHandler messageHandler = task -> {
        RemoteCommand command = task.getCommand();
        String content = command.getContent();
        RemoteCommand response = new RemoteCommand();
        try {
            ImMessage message = JsonUtil.jsonToPojo(content, ImMessage.class);
            WindowManager.forwardMessage(message);
        } catch (Exception e) {
            log.warn("message forward failed, command: {}", command.toString());
            response.setEventType(EventType.OK);
            response.setEventType(EventType.FAILED);
        }
        return response;
    };

    private static final CommandHandler notifyHandler = task -> {
        Set<ServerNode> remoteNodes = LocalCacheNodeTable.getRemoteNodes();
        remoteNodes.forEach(node -> {
            try {
                NettyRemotingClient.instance().oneWaySend(task.getCommand(), node, 30);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        return null;
    };

    static {
        handlerMap = new ConcurrentHashMap<>();
//        handlerMap.put(EventType.CLOSE_WINDOW, null);
        handlerMap.put(EventType.NEW_WINDOW, newWindowHandler);
        handlerMap.put(EventType.MESSAGE, messageHandler);
        handlerMap.put(EventType.NOTIFY, notifyHandler);
    }

    public CommandDispatcher() {
        taskExecutor = new ThreadPoolExecutor(
                10,
                10,
                0,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(1000)
        );
        loopThread = Executors.newSingleThreadExecutor();
        start();
    }

    public static CommandDispatcher getInstance() {
        return instance;
    }

    public static void doDispatcher(CommandTask task) {
        CommandHandler handler = handlerMap.get(task.getEventType());
        task.setHandler(handler);
        instance.taskExecutor.submit(task);
    }

    public void start() {
        loopThread.submit(() -> {
            while (true) {
                CommandTask task = null;
                try {
                    task = commandQueue.pop();
                    CommandHandler handler = handlerMap.get(task.getEventType());
                    task.setHandler(handler);
                    taskExecutor.submit(task);
                } catch (Exception e) {
                    log.error("process command failed", e);
                }
            }
        });
    }

    private void doRequestCommand(CommandTask task) {
        CommandHandler handler = handlerMap.get(task.getEventType());
        RemoteCommand request = task.getCommand();
        CommandCallback callback = (response) -> {
            response.setRequestType(RequestType.RESPONSE);
            response.setCommandId(request.getCommandId());
            try {
                task.getCtx().writeAndFlush(response);
            } catch (Exception e) {
                log.error("process request over, but response failed", e);
                log.error(request.toString());
                log.error(response.toString());
            }
        };
        task.setHandler(handler);
        task.setCallback(callback);
        taskExecutor.submit(task);
    }

    private void doResponseCommand(CommandTask task) {
        ResponseFuture responseFuture = task.getResponseFuture();
        RemoteCommand response = task.getCommand();
        if (responseFuture != null) {
            responseFuture.setResponse(response);
            if (responseFuture.getCallback() != null) {
                responseFuture.executeCallback();
            } else {
                responseFuture.putResponse(response);
                responseFuture.release();
            }
        } else {
            log.warn("receive response, but not matched any request");
            log.warn(response.toString());
        }
    }


}
